package rpc;

import java.util.UUID;  

import com.rabbitmq.client.AMQP.BasicProperties;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.QueueingConsumer;  

public class RPCClient2 {
	private Connection connection;  
    private Channel channel;  
    private String requestQueueName = "rpc_queue";  
    private String replyQueueName;  
    private QueueingConsumer consumer;  
  
    public RPCClient2() throws Exception {  
        ConnectionFactory factory = new ConnectionFactory();  
        factory.setHost("10.20.53.21");  
        connection = factory.newConnection();  
        channel = connection.createChannel();  
  
        replyQueueName = channel.queueDeclare().getQueue();  
        consumer = new QueueingConsumer(channel);  
        channel.basicConsume(replyQueueName, true, consumer);  
    }  
   
	public String call(String message) throws Exception {  
        String response = null;  
        String corrId = UUID.randomUUID().toString();  
  
        BasicProperties props = new BasicProperties();  
        props.setReplyTo(replyQueueName);  
        props.setCorrelationId(corrId);  
  
        channel.basicPublish("", requestQueueName, props, message.getBytes());  
        System.out.println("call msg[ " + message + " ], corrId[ "+corrId+"]" );  
        while (true) {  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {  
                response = new String(delivery.getBody(), "UTF-8");  
                break;  
            }  
        }  
  
        return response;  
    }
	public String send(String message) throws Exception {  
        String corrId = UUID.randomUUID().toString();  
  
        BasicProperties props = new BasicProperties();  
        props.setReplyTo(replyQueueName);  
        props.setCorrelationId(corrId);  
  
        channel.basicPublish("", requestQueueName, props, message.getBytes());  
        System.out.println("send msg[ " + message + " ], reply Q["+replyQueueName+"], corrId[ "+corrId+"]" );  
        
        return corrId;  
    }
	
	public String getmsgBycorrId(String corrId) throws Exception {  
        String response = null;   
        String curCorrID = null;
        System.out.println("getmsgBycorrId corrId[ "+corrId+"]" );  
        while (true) {  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
            curCorrID = delivery.getProperties().getCorrelationId();
            System.out.println("getmsgBycorrId curCorrID[ "+curCorrID+"]" );  
            if (curCorrID.equals(corrId)) {  
                response = new String(delivery.getBody(), "UTF-8");  
                break;  
            }  
        }  
  
        return response;  
    }	
  
    public void close() throws Exception {  
        connection.close();  
    }  
    
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		RPCClient2 fibonacciRpc = null;  
        String response = null;  
        try {  
            fibonacciRpc = new RPCClient2();  
//	          System.out.println(" [x] try send (30)");  
//	          response = fibonacciRpc.call("30");
	          String corrId1 = fibonacciRpc.send("5");
	          String corrId2 = fibonacciRpc.send("3");
	          String corrId3 = fibonacciRpc.send("4");
	          String corrId4 = fibonacciRpc.send("7"); 
	          
	          System.out.println("response2 ["+fibonacciRpc.getmsgBycorrId(corrId2)+"]");
	          System.out.println("response1 ["+fibonacciRpc.getmsgBycorrId(corrId1)+"]");
	          
	          System.out.println("response3 ["+fibonacciRpc.getmsgBycorrId(corrId3)+"]");
	          System.out.println("response4 ["+fibonacciRpc.getmsgBycorrId(corrId4)+"]");
//            System.out.println(" [x] Requesting fib(30)");  
//            response = fibonacciRpc.call("30");  
//            System.out.println(" [.] Got '" + response + "'");  
//            System.out.println(" [x] Requesting fib(-1)");  
//            response = fibonacciRpc.call("-1");  
//            System.out.println(" [.] Got '" + response + "'");  
//            System.out.println(" [x] Requesting fib(a)");  
//            response = fibonacciRpc.call("a");  
//            System.out.println(" [.] Got '" + response + "'");  
        } catch (Exception e) {  
            e.printStackTrace();  
        } finally {  
            if (fibonacciRpc != null) {  
                try {  
                    fibonacciRpc.close();  
                } catch (Exception ignore) {  
                }  
            }  
        }  
	}

}
